package com.ybzl.gateway.utils;

import com.ybzl.gateway.domain.NoIbeacon;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class SlidingWindowProcessor {
    // FIFO
    private static final ConcurrentLinkedQueue<NoIbeacon> window = new ConcurrentLinkedQueue<>();

    @Autowired
    private NoIbeaconCalc ibeaconCalc;

    // 添加队列监控
    private final AtomicInteger queueSize = new AtomicInteger(0);

    private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
        private final AtomicInteger threadCount = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            // 自定义线程名称格式
            Thread thread = new Thread(r, "push-data-" + threadCount.getAndIncrement());
            // 可选的额外配置
            thread.setDaemon(false); // 设置为非守护线程
            thread.setPriority(Thread.NORM_PRIORITY);
            return thread;
        }
    });


    /*@PostConstruct
    public void t() {
        scheduler.scheduleAtFixedRate(() -> {
            try {
                if (!window.isEmpty()) {
                    log.info("Queue poll Before Size is:{}", window.size());
                    // 一次性取出10条记录
                    List<NoIbeacon> firstTen = new ArrayList<>();
                    for (int i = 0; i < 500 && !window.isEmpty(); i++) {
                        NoIbeacon b = window.peek(); // 仅查看队首元素不移除 // 【非原子操作】
                        if (!ObjectUtils.isEmpty(b)) {
                            firstTen.add(window.poll()); // 移除并返回队首元素 【另一个非原子操作】
                        }
                    }
                    log.info("Queue poll After Size is:{}", window.size());
                    ibeaconCalc.getNoIbeacon(firstTen);
                }

            } catch (Exception e) {
                log.error("间歇性取值异常.{}", e.getMessage(), e);
            }

        }, 0, 1, TimeUnit.SECONDS);

    }*/
    @PostConstruct
    public void initProcessor() {
        scheduler.scheduleAtFixedRate(() -> {
            try {
                int currentSize = window.size();
                log.info("Queue poll Before Size is:{}", currentSize);
                if (currentSize == 0) return;

                // 原子获取批量数据
                List<NoIbeacon> batch = new ArrayList<>(Math.min(500, currentSize));
                for (int i = 0; i < 500 && !window.isEmpty(); i++) {
                    NoIbeacon data = window.poll();
                    if (data != null) {
                        batch.add(data);
                        queueSize.decrementAndGet();
                    }
                }
                log.info("Queue poll After Size is:{}", window.size());
                if (!batch.isEmpty()) {
                    ibeaconCalc.processBatch(batch);
                }
            } catch (Exception e) {
                log.error("处理异常", e);
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    public void addData(NoIbeacon data) {
        if (data == null || data.getTs() == null) return;

        int currentSize = queueSize.incrementAndGet();
        if (currentSize > 9000) {
            log.warn("队列接近满载: {}/10000", currentSize);
        }
        window.add(data);
    }

}


